package learn

import domain.Domain.Access
import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.operators.StreamMap
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object test01Transform {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(2)
        //env.socketTextStream("hadoop000", 9527)
        //  .transform("map", new StreamMap(new MapFunction[String, (String, Int)] {
        //      override def map(value: String): (String, Int) = {
        //          val split = value.split(",")
        //          (split(0), 1)
        //      }
        //  })).print()

        env.socketTextStream("hadoop000", 9527).map(new MapFunc).print()

        env.execute("stageTest")
    }
}

class MapFunc extends RichMapFunction[String, Access] {
    override def open(parameters: Configuration): Unit = {
        println("======open=====")
    }

    override def map(value: String): Access = {
        Access(value, value)
    }

    override def close(): Unit = {

    }
}
